/*
 * REFTaskIOEventDispatcher.cpp
 *
 *  Created on: Jun 5, 2013
 *      Author: Lucifer
 */

#include "../pch.h"
#include "REFIOEventContext.h"
#include "REFIOEventDispatcher.h"
#include "REFIOEventService.h"
#include <errno.h>

namespace BFX
{

#ifndef	BFX_IO_HAS_IOCP

////////////////////////////////////////////////////////////////////////////////
// IOEventDispatchQueue

/**
 * Initializes a new instance of BFX::IOEventDispatchQueue.
 */
IOEventDispatchQueue::IOEventDispatchQueue()
{
}

/**
 * Destroys the BFX::IOEventDispatchQueue instance.
 */
IOEventDispatchQueue::~IOEventDispatchQueue()
{
}

/**
 * Add a new operation to the queue.
 *
 * @param fd The file descriptor.
 * @param pContext The event context associated with the given file descriptor.
 * @return Returns true if this is the only operation for the given file descriptor. In this case the event
 * 		demultiplexing function my need to restarted.
 */
bool IOEventDispatchQueue::WaitEvent(SOCKET fd, REFIOEventContext* pContext)
{
	BFX_REQUIRE(fd != INVALID_SOCKET);
	BFX_REQUIRE(pContext != NULL);

	ContextQueue& queue = m_contexts[fd];	// NOTE: this operation will create the queue object explicitly.
	bool bFirst = queue.IsEmpty();
	queue.Enqueue(pContext);
	return bFirst;
}

/**
 * Cancel all operations associated with the descriptor. Any operations pending for file descriptor will be
 * notified that they have been canceled.
 *
 * @param[in] fd The file descriptor.
 * @param[out] pContexts The queue uses to receive contexts were canaled.
 * @param[in] nErrorCode The system error code on this cancel operation. The default specifying value (0) indicates
 * 		canceled by user-aborted.
 * @return True if any operations were canceled. In this case the event demultiplexing function may need to restarted.
 */
bool IOEventDispatchQueue::StopWaitEvents(SOCKET fd, REFIOEventContextQueue* pContexts, int nErrorCode)
{
	BFX_REQUIRE(fd != INVALID_SOCKET);
	BFX_REQUIRE(pContexts != NULL);

	if (nErrorCode == 0)
		nErrorCode =
#ifdef	_WIN32
				WSAECONNABORTED;
#else
				ECONNABORTED;
#endif

	ContextMap::POSITION entry = m_contexts.LookupEntry(fd);
	if (entry != NULL)
	{
		ContextQueue& queue = entry->value;
		while (!queue.IsEmpty())
		{
			REF<REFIOEventContext> pContext = queue.Dequeue();
			pContext->SetErrorCode(nErrorCode);
			pContexts->Enqueue(pContext);
		}
		m_contexts.Remove(fd);
		return true;
	}
	return false;
}

/**
 * Fill a descriptor set with the descriptors corresponding to each active contexts.
 *
 * @param pFdSet[in,out] The file descriptor set.
 * @param pContexts[out] The queue stores contexts only when descriptors fail to be added to the descriptor set.
 */
void IOEventDispatchQueue::GetFileDescriptors(REFFileDescriptorSet* pFdSet, REFIOEventContextQueue* pContexts)
{
	BFX_REQUIRE(pFdSet != NULL);
	BFX_REQUIRE(pContexts != NULL);

	for (ContextMap::POSITION entry = m_contexts.GetFirstEntry(); entry != NULL;
			entry = m_contexts.GetNextEntry(entry))
	{
		SOCKET fd = entry->key;
		if (!pFdSet->Set(fd))
		{
			// fd_set is full.
			int nErrorCode =
#ifdef	_WIN32
					WSAENOBUFS;
#else
					ENOBUFS;
#endif
			StopWaitEvents(fd, pContexts, nErrorCode);
		}
	}
}

void IOEventDispatchQueue::Clear(REFIOEventContextQueue* pContexts)
{
	for (ContextMap::POSITION entry = m_contexts.GetFirstEntry(); entry != NULL;)
	{
		SOCKET fd = entry->key;
		ContextQueue& queue = entry->value;
		entry = m_contexts.GetNextEntry(entry);
		if (pContexts != NULL)
		{
			while (!queue.IsEmpty())
			{
				pContexts->Enqueue(queue.Dequeue());
			}
		}
		m_contexts.Remove(fd);
	}
}

/**
 * Perform all operations associated with the given file descriptor.
 *
 * @param pFdSet[in] The file descriptor set.
 * @param pOperations[out] The queue to receive operations will be completed.
 */
void IOEventDispatchQueue::PerformEvents(REFFileDescriptorSet* pFdSet, REFIOEventContextQueue* pContexts)
{
	BFX_ENSURE(pFdSet != NULL);
	BFX_ENSURE(pContexts != NULL);

	for (ContextMap::POSITION entry = m_contexts.GetFirstEntry(); entry != NULL;)
	{
		SOCKET fd = entry->key;
		ContextQueue& queue = entry->value;
		if (pFdSet->IsSet(fd))
		{
			while (!queue.IsEmpty())
			{
				// attempt to perform the first handler.
				REFIOEventContext* pContext = queue.Peek();
				if (pContext->Perform())
				{
					// only dequeue in case of success.
					pContexts->Enqueue(pContext);
					queue.Dequeue();
				}
				else
					break;	// there are still unfinished operations queued for the descriptor.
			}
		}
		entry = m_contexts.GetNextEntry(entry);
		if (queue.IsEmpty())
			m_contexts.Remove(fd);
	}
}

bool IOEventDispatchQueue::IsEmpty() const
{
	return m_contexts.IsEmpty();
}

////////////////////////////////////////////////////////////////////////////////
// REFIOEventDispatcher

#if defined(BFX_IO_HAS_EPOLL)

#elif defined(BFX_SIO_HAS_KQUEUE)

#elif defined(BFX_IO_HAS_DEV_POLL)

#else	//	BFX_IO_HAS_SELECT

/**
 * Initializes a new instance of BFX::REFSelectDispatcher.
 * @param[in] bOwnThread Uses own dispatch thread.
 */
REFIOEventDispatcher::REFIOEventDispatcher(REFIOEventService* pIOService)
{
	m_pIOService = pIOService;
	m_bShutdown = false;

	m_pIOService->InitializeEventDispatcher();
}

/**
 * Destroys the BFX::REFSelectDispatcher instance.
 */
REFIOEventDispatcher::~REFIOEventDispatcher()
{
	Shutdown();
}

/**
 * Register the file descriptor with the dispatcher.
 *
 * @param fd The file descriptor.
 * @return True on success; otherwise false.
 */
bool REFIOEventDispatcher::RegisterFileDescriptor(SOCKET fd)
{
	return true;	// NOTE: intentionally do nothing.
}

/**
 * Unregister the file descriptor.
 * This operation will cancel all waiting event contexts associated with the given descriptor.
 *
 * @param fd The file descriptor.
 */
void REFIOEventDispatcher::UnregisterFileDescriptor(SOCKET fd)
{
	StopWaitEvents(fd);
}

/**
 * Starts a new operation.
 * The operation will be performed when the given descriptor is flagged as ready, or an error has occurred.
 *
 * @param fd The file descriptor.
 * @param pContext The event context associated to the given descriptor.
 * @param event The event type.
 */
void REFIOEventDispatcher::WaitEvent(SOCKET fd, REFIOEventContext* pContext, EVENT_TYPE event)
{
	BFX_REQUIRE(event >= 0 && event < MAX_EVENTS);

	MutexHolder lock(&m_mutex);
	if (m_bShutdown)
	{
		m_pIOService->PostImmediateCompletion(pContext);
		return;
	}

	m_pIOService->WorkStarted();
	bool bFirst = m_queue[event].WaitEvent(fd, pContext);
	if (bFirst)
		m_interrupter.Interrupt();
}

/**
 * Run select once until interrupted or events are ready to be dispatched.
 *
 * @param[in] bBlocking The value that indicates whether the event-loop is in blocking mode.
 * @param[out] pContexts The queue to receive contexts will be completed.
 */
void REFIOEventDispatcher::DispatchEvents(bool bBlocking, REFIOEventContextQueue* pContexts)
{
	MutexHolder lock(&m_mutex);

	// set up the descriptor sets.
	REFFileDescriptorSet fdSets[MAX_SELECT_EVENTS];
	SOCKET fdMax = 0;
	bool bNotingToDo = true;

	// install our interrupter.
	fdSets[READ_EVENT].Set(m_interrupter.GetFileDescriptor());

	for (int i = 0; i < MAX_SELECT_EVENTS; i ++)
	{
		fdSets[i].AddRef();	// NOTE: prevent external call attempts to delete on stack variables.

		bNotingToDo = (bNotingToDo && m_queue[i].IsEmpty());
		m_queue[i].GetFileDescriptors(&fdSets[i], pContexts);
		fdMax = __MAX(fdSets[i].GetMaxFileDescriptor(), fdMax);
	}

#if defined(_WIN32) || defined(__CYGWIN__)
	// connection operations on Windows use both except and write fd_sets.
	bNotingToDo = (bNotingToDo && m_queue[CONNECT_EVENT].IsEmpty());
	m_queue[CONNECT_EVENT].GetFileDescriptors(&fdSets[WRITE_EVENT], pContexts);
	fdMax = __MAX(fdSets[WRITE_EVENT].GetMaxFileDescriptor(), fdMax);
	m_queue[CONNECT_EVENT].GetFileDescriptors(&fdSets[EXCEPT_EVENT], pContexts);
	fdMax = __MAX(fdSets[EXCEPT_EVENT].GetMaxFileDescriptor(), fdMax);
#endif

	// we can return immediately if there's no work to do and the reactor is not supposed to block.
	if (!bBlocking && bNotingToDo)
		return;

	lock.Release();

	// determine how long to block while waiting for events.
	timeval timeout = { 0, 0 };
	if (bBlocking)
	{
		const int US = 5 * 60 * 1000000;	// we will wait no longer than 5 minutes
		timeout.tv_sec = US / 1000000;
		timeout.tv_usec = US % 1000000;
	}

	// block on the select call until at least one of descriptors become ready.
	int nRetVal = select(fdMax + 1, fdSets[READ_EVENT], fdSets[WRITE_EVENT], fdSets[EXCEPT_EVENT], &timeout);

	// reset our interrupter.
    if (nRetVal > 0 && fdSets[READ_EVENT].IsSet(m_interrupter.GetFileDescriptor()))
    	m_interrupter.Reset();

	lock.Acquire();

	// dispatch all ready operations.
	if (nRetVal > 0)
	{
#if defined(_WIN32) || defined(__CYGWIN__)
		// connection operations on Windows use both except and write fd_sets.
		m_queue[CONNECT_EVENT].PerformEvents(&fdSets[WRITE_EVENT], pContexts);
		m_queue[CONNECT_EVENT].PerformEvents(&fdSets[EXCEPT_EVENT], pContexts);
#endif
		// exception operations must be processed first to ensure that any out-of-band data is read before normal data.
		for (int i = MAX_SELECT_EVENTS - 1; i >= 0; --i)
			m_queue[i].PerformEvents(&fdSets[i], pContexts);
	}
}

/**
 * Shutdown the dispatcher.
 * Destroys all user defined objects.
 */
void REFIOEventDispatcher::Shutdown()
{
	m_bShutdown = true;

	// removes all remaining tasks.
	REF<REFIOEventContextQueue> pContexts = new REFIOEventContextQueue();
	for (int i = 0; i < MAX_EVENTS; i ++)
	{
		m_queue[i].Clear(pContexts);
	}

	// TODO: waiting timers.
}

/**
 * Cancel all operations associated with the given descriptor. The contexts associated with the
 * file descriptor will be invoked with given error code.
 *
 * @param[in] fd The file descriptor.
 * @param[in] nErrorCode The system error code on this cancel operation. The default specifying value (0) indicates
 * 		canceled by user aborted.
 */
void REFIOEventDispatcher::StopWaitEvents(SOCKET fd, int nErrorCode)
{
	MutexHolder lock(&m_mutex);

	bool bNeedInterrupt = false;
	REF<REFIOEventContextQueue> pContexts = new REFIOEventContextQueue();
	for (int i = 0; i < MAX_EVENTS; i++)
		bNeedInterrupt = (bNeedInterrupt || m_queue[i].StopWaitEvents(fd, pContexts, nErrorCode));

	while (!pContexts->IsEmpty())
	{
		REF<REFIOEventContext> pContext = pContexts->Dequeue();
		m_pIOService->PostDeferredCompletion(pContext);
	}

	if (bNeedInterrupt)
		m_interrupter.Interrupt();
}

#endif	//	BFX_IO_HAS_SELECT

#endif	//	!BFX_IO_HAS_IOCP

} /* namespace BFX */
